package com.gis.aoitest.source;

import com.gis.aoitest.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Random;

/**
 * @author LnnuUser
 * @create 2021-09-06-下午9:22
 */
public class SourceTest04_UDF {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 从自定义source读取数据
        DataStreamSource<SensorReading> dataStream = env.addSource(new MySensorSource());

        dataStream.print();

        env.execute();
    }

    // 实现自定义类MySensorSource,实现SourceFunction接口
    private static class MySensorSource  implements SourceFunction<SensorReading> {

        private boolean flg = true;

        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception {

            // 定义一个随机数发生器
            Random random = new Random();

            // 设置10个传感器的初始温度
            HashMap<String, Double> sensorTempMap = new HashMap<>();
            for (int i = 0; i < 10; i++) {
                sensorTempMap.put("sensor_" + (i + 1), 60 + random.nextGaussian() * 20);
            }
            while (flg) {
                for (String sensorId: sensorTempMap.keySet()) {
                    // 在当前温度基础上，随机波动
                    Double newtemp = sensorTempMap.get(sensorId) + random.nextGaussian();
                    sensorTempMap.put(sensorId, newtemp);
                    ctx.collect(new SensorReading(sensorId, System.currentTimeMillis(), newtemp));
                }

                // 控制输出频率
                Thread.sleep(1000L);

            }
        }

        @Override
        public void cancel() {
            flg = false;
        }
    }
}
